package com.thb.pay.mq.process;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: lucky
 * @Descrpiton: 多线程执行器
 * @Date: 9:44 2018/12/4
 * @Modify by:
 */
@Slf4j
public class PoolExecutorProcess<T> {

    // 线程池维护线程的最少数量
    private final static int CORE_POOL_SIZE = 2;//核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）//超过这个限制 将不做拒绝加入到本线程池中
    // 线程池维护线程的最大数量
    private final static int MAX_POOL_SIZE = 20;
    // 线程池维护线程所允许的空闲时间
    private final static int KEEP_ALIVE_TIME = 0;
    // 线程池所使用的缓冲队列大小
    private final static int WORK_QUEUE_SIZE = 50;
    //用于从IOC里取对象
    private BeanFactory factory; //如果实现Runnable的类是通过spring的application.xml文件进行注入,可通过 factory.getBean()获取，这里只是提一下
    private String queueName;
    private ThreadPoolExecutor threadPoolExecutor;
    //执行器线程池的状态
    private volatile Integer state = 0;//0 ide  1 running
    private final RejectedExecutionHandler handler = ((runnable, executor) -> {
        this.state = 1;//当线程池的容量满了 无法再次进去线程池进行消费[可扩展成是否加入到任务缓冲队列里面]

    });
    private MessageProcess<T> messageProcess;

    private Environment environment;

    public PoolExecutorProcess(Environment environment, MessageProcess<T> messageProcess, String queueName) {
        this.environment = environment;
        this.messageProcess = messageProcess;
        this.queueName = queueName;
        threadPoolExecutor = new ThreadPoolExecutor(getThreadCount(), MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue(WORK_QUEUE_SIZE), this.handler);
    }

    protected int getThreadCount() {
        int targetThreadCount = CORE_POOL_SIZE;
        String threadCount;
        threadCount = environment.getProperty("queue.comm.threadCount");//公共默认配置大小
        if (!StringUtils.isEmpty(threadCount)) {
            targetThreadCount = Integer.parseInt(threadCount);
            if (targetThreadCount > MAX_POOL_SIZE) {
                targetThreadCount = MAX_POOL_SIZE;
            }
            return targetThreadCount;
        }
        if (!StringUtils.isEmpty(queueName)) {
            threadCount = environment.getProperty("queue." + queueName + ".threadCount");//公共默认配置大小
        }
        if (!StringUtils.isEmpty(threadCount)) {
            targetThreadCount = Integer.parseInt(threadCount);
        }
        if (targetThreadCount > MAX_POOL_SIZE) {
            targetThreadCount = MAX_POOL_SIZE;
        }
        if (targetThreadCount <= 0) {
            targetThreadCount = CORE_POOL_SIZE;
        }
        return targetThreadCount;
    }

    public boolean checkPoolIde() {

        if (threadPoolExecutor.getQueue() != null && threadPoolExecutor.getQueue().size() >= MAX_POOL_SIZE) {
            log.info("threadPoolExecutor.getQueue().size()" + threadPoolExecutor.getQueue().size());
            this.state = 1;
            return false;
        } else {
            this.state = 0;//空闲
            return true;
        }
    }


    public void process(T data) {
        if (checkPoolIde()) {//空闲才能将数据放入线程池执行
            ProcessTask<T> processTask = new ProcessTask(data, messageProcess);
            threadPoolExecutor.submit(processTask);

        }

    }


    //终止数据线程池
    @PreDestroy
    public void shutdown() {
        threadPoolExecutor.shutdown();

    }


    public MessageProcess<T> getMessageProcess() {
        return messageProcess;
    }

    public void setMessageProcess(MessageProcess<T> messageProcess) {
        this.messageProcess = messageProcess;
    }
}
